package com.navinfo.opentsp.platform.computing.serializer;

import com.navinfo.opentsp.platform.location.protocol.common.LCLocationData;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.conf.ComponentConfiguration;
import org.apache.flume.sink.hbase.HbaseEventSerializer;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Row;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 位置信息LocationData序列化
 */
public class HbaseLocationDataSerializer implements HbaseEventSerializer {
    private static final Logger log = LoggerFactory.getLogger(HbaseLocationDataSerializer.class);
    private Event currentEvent;
    private byte[] colFam;

    public void initialize(Event event, byte[] colFam) {
        //byte[]字节型数组
        this.currentEvent = event;
        this.colFam = colFam;
    }
    public void close() {
        this.colFam = null;
        this.currentEvent = null;
    }

    public void configure(Context context) {   }
    public void configure(ComponentConfiguration conf) {    }

    public List<Increment> getIncrements() {
        return  new ArrayList<Increment>();
    }

    public List<Row> getActions() {
        try {
            LCLocationData.LocationData locationData = LCLocationData.LocationData.parseFrom(currentEvent.getBody());

            //rowkey:reverse(tId)|3000000000L-gpsTime
            String rowKey = new StringBuilder(Long.valueOf(locationData.getTerminalId()).toString()).reverse() +"|" + (3000000000L-locationData.getGpsDate());
            List<Row> puts = new ArrayList<Row>();
            puts.add(preparedata(new Put(rowKey.getBytes()), locationData));
            return puts;
        } catch (Exception e) {
            log.error("HBase落盘序列化失败："+e.getMessage(), e);
        }

        return  Collections.emptyList();
    }

    private  Put preparedata(Put putReq, LCLocationData.LocationData locationData) throws Exception {
        putReq.addColumn(colFam, Bytes.toBytes("alarm"),Bytes.toBytes(String.valueOf(locationData.getAlarm())));
        putReq.addColumn(colFam, Bytes.toBytes("status"),Bytes.toBytes(String.valueOf(locationData.getStatus())));
        putReq.addColumn(colFam, Bytes.toBytes("longitude"),Bytes.toBytes(String.valueOf(locationData.getLongitude())));
        putReq.addColumn(colFam, Bytes.toBytes("latitude"),Bytes.toBytes(String.valueOf(locationData.getLatitude())));
        putReq.addColumn(colFam, Bytes.toBytes("originalLng"),Bytes.toBytes(String.valueOf(locationData.getOriginalLng())));
        putReq.addColumn(colFam, Bytes.toBytes("originalLat"),Bytes.toBytes(String.valueOf(locationData.getOriginalLat())));
        putReq.addColumn(colFam, Bytes.toBytes("height"),Bytes.toBytes(String.valueOf(locationData.getHeight())));
        putReq.addColumn(colFam, Bytes.toBytes("speed"),Bytes.toBytes(String.valueOf(locationData.getSpeed())));
        putReq.addColumn(colFam, Bytes.toBytes("direction"),Bytes.toBytes(String.valueOf(locationData.getDirection())));
        putReq.addColumn(colFam, Bytes.toBytes("gpsDate"),Bytes.toBytes(String.valueOf(locationData.getGpsDate())));
        putReq.addColumn(colFam, Bytes.toBytes("mileage"),Bytes.toBytes(String.valueOf(locationData.getMileage())));
        putReq.addColumn(colFam, Bytes.toBytes("oil"),Bytes.toBytes(String.valueOf(locationData.getOil())));
        putReq.addColumn(colFam, Bytes.toBytes("recorderSpeed"),Bytes.toBytes(String.valueOf(locationData.getRecorderSpeed())));
        putReq.addColumn(colFam, Bytes.toBytes("standardMileage"),Bytes.toBytes(String.valueOf(locationData.getStandardMileage())));
        putReq.addColumn(colFam, Bytes.toBytes("standardFuelCon"),Bytes.toBytes(String.valueOf(locationData.getStandardFuelCon())));
        putReq.addColumn(colFam, Bytes.toBytes("originalPB"),locationData.toByteArray());

        return putReq;
    }
}
